package com.atguigu.realtime.util;

import com.atguigu.realtime.common.Constant;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @Author lzc
 * @Date 2023/3/8 14:25
 */
public class FlinkSourceUtil {
    
    public static KafkaSource<String> getKafkaSource(String groupId, String topic) {
        return KafkaSource.<String>builder()
            .setBootstrapServers(Constant.KAFKA_SERVERS)
            .setGroupId(groupId)
            .setTopics(topic)
            // 首次消费从最新的地方开始消费. 如果是从 savepoint 或者 checkpoint 恢复,则从状态中的记录偏移量开始消费
            .setStartingOffsets(OffsetsInitializer.latest())
            // 设置读取消息的隔离级别:只读已提交的数据
            .setProperty("isolation.level", "read_committed")
            // 设置反序列化器
            .setValueOnlyDeserializer(new DeserializationSchema<String>() { // 自定义反序列化
                
                // 反序列化
                @Override
                public String deserialize(byte[] message) throws IOException {
                    if (message != null) {
                        return new String(message, StandardCharsets.UTF_8);
                    }
                    return null;  // 如果来的是 null, 写向流中写入一个 null, flink 会自动的忽略这个值
                }
                
                // 是否结束流
                @Override
                public boolean isEndOfStream(String nextElement) {
                    return false;  // 流永远不结束
                }
                
                // 定义结果类型
                @Override
                public TypeInformation<String> getProducedType() {
                    //                    return Types.STRING;  // 当是一些简单类型: 基本类型, 字符串, 元组
                    //                    return TypeInformation.of(String.class);  // 当是自定义类型
                    return TypeInformation.of(new TypeHint<String>() {});
                    
                    
                }
            })  // SimpleStringSchema 再反序列化 null 时候会报错: 空指针
            .build();
    }
}
